package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_每分钟每类页面访问人数TOP10 {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中dwd层的行为日志明细
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        tenv.executeSql(
                "CREATE TABLE dash_03_page_top10  (                                         \n" +
                        "  page_type    string                                              \n" +
                        "  ,rn           bigint                                             \n" +
                        "  ,window_start timestamp(3)                                        \n" +
                        "  ,window_end   timestamp(3)                                       \n" +
                        "  ,page_url     string                                             \n" +
                        "  ,page_uv      bigint                                             \n" +
                        "  ,PRIMARY KEY (page_type,rn) NOT ENFORCED                        \n" +
                        ") with (                                                          \n" +
                        "   'connector' = 'jdbc',                                          \n" +
                        "   'url' = 'jdbc:mysql://doitedu:3306/doit43',                    \n" +
                        "   'table-name' = 'dash_03_type_topn_page',                        \n" +
                        "   'username' = 'root'      ,                                     \n" +
                        "   'password' = 'root'                                            \n" +
                        ")                                                                 "
        );



        // 写统计sql
        // 截止到此刻，今天总的pv数，uv数，会话数，并且每1分钟更新一次结果
        tenv.executeSql(
                        " insert into dash_03_page_top10  " +
                        " WITH tmp AS (\n" +
                                "SELECT\n" +
                                "    user_id,\n" +
                                "    page_type,\n" +
                                "    properties['url'] as page_url,\n" +
                                "    rt \n" +
                                "FROM dwd_events_kafka\n" +
                                "WHERE event_id = 'page_load' )\n" +
                                "\n" +
                                "SELECT \n" +
                                "    *  \n" +
                                "FROM (\n" +
                                "    SELECT\n" +
                                "      page_type,   \n" +
                                "      ROW_NUMBER() OVER(PARTITION BY window_start,window_end,page_type ORDER BY page_uv DESC) AS rn," +
                                "      window_start,\n" +
                                "      window_end,\n" +
                                "      page_url,\n" +
                                "      page_uv \n" +
                                "    FROM (\n" +
                                "        SELECT\n" +
                                "            window_start,\n" +
                                "        \twindow_end,\n" +
                                "        \tpage_type,\n" +
                                "        \tpage_url,\n" +
                                "        \tcount(distinct user_id) as page_uv\n" +
                                "        FROM  TABLE(\n" +
                                "            TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)\n" +
                                "        )\n" +
                                "        GROUP BY \n" +
                                "            window_start,\n" +
                                "        \twindow_end,\n" +
                                "        \tpage_type,\n" +
                                "        \tpage_url\n" +
                                "    )\t\n" +
                                ")\n" +
                                "WHERE rn<=10"
        );



    }

}
